#!/usr/bin/env python3
"""
Requirements.txt 诊断脚本
系统性分析requirements.txt文件中的潜在问题
"""

import re
import sys
from pathlib import Path
from collections import defaultdict, Counter
from typing import Dict, List, Tuple, Set
import subprocess
import json

class RequirementsDiagnostic:
    def __init__(self, requirements_file: str = "requirements.txt"):
        self.requirements_file = Path(requirements_file)
        self.issues = []
        self.warnings = []
        self.info = []
        
    def log_issue(self, category: str, message: str, line_num: int = None):
        """记录问题"""
        issue = {"category": category, "message": message, "severity": "ERROR"}
        if line_num is not None:
            issue["line"] = str(line_num)
        self.issues.append(issue)
        print(f"❌ [ERROR] {category}: {message}")
        
    def log_warning(self, category: str, message: str, line_num: int = None):
        """记录警告"""
        warning = {"category": category, "message": message, "severity": "WARNING"}
        if line_num is not None:
            warning["line"] = str(line_num)
        self.warnings.append(warning)
        print(f"⚠️  [WARNING] {category}: {message}")
        
    def log_info(self, category: str, message: str):
        """记录信息"""
        info = {"category": category, "message": message, "severity": "INFO"}
        self.info.append(info)
        print(f"ℹ️  [INFO] {category}: {message}")

    def parse_requirements(self) -> List[Dict]:
        """解析requirements文件"""
        if not self.requirements_file.exists():
            self.log_issue("FILE_NOT_FOUND", f"文件不存在: {self.requirements_file}")
            return []
            
        requirements = []
        try:
            with open(self.requirements_file, 'r', encoding='utf-8') as f:
                lines = f.readlines()
                
            for line_num, line in enumerate(lines, 1):
                line = line.strip()
                if not line or line.startswith('#'):
                    continue
                    
                # 解析依赖行
                req_info = self._parse_requirement_line(line, line_num)
                if req_info and req_info.get("package"):
                    requirements.append(req_info)
                    
        except Exception as e:
            self.log_issue("PARSE_ERROR", f"解析文件失败: {str(e)}")
            
        return requirements

    def _parse_requirement_line(self, line: str, line_num: int) -> Dict:
        """解析单个依赖行"""
        # 基本的requirements.txt格式解析
        # 支持: package==1.0.0, package>=1.0.0,<2.0.0, package[extra]==1.0.0
        
        # 移除注释
        if '#' in line:
            line = line.split('#')[0].strip()
            
        if not line:
            return {}
            
        # 解析包名和版本约束
        match = re.match(r'^([a-zA-Z0-9_.-]+(?:\[[^\]]+\])?)\s*([><=!~]+.*)?$', line)
        if not match:
            self.log_warning("PARSE_WARNING", f"无法解析依赖行: {line}", line_num)
            return {}
            
        package_name = match.group(1)
        version_spec = match.group(2) or ""
        
        # 提取基础包名（移除extras）
        base_package = re.sub(r'\[.*\]', '', package_name)
        
        return {
            "line_num": line_num,
            "raw_line": line,
            "package": base_package,
            "full_package": package_name,
            "version_spec": version_spec.strip(),
            "is_pinned": "==" in version_spec,
            "is_range": any(op in version_spec for op in [">=", "<=", ">", "<", "~="])
        }

    def check_duplicate_dependencies(self, requirements: List[Dict]):
        """检查重复依赖 - 问题源1"""
        print("\n🔍 检查重复依赖...")
        
        package_occurrences = defaultdict(list)
        for req in requirements:
            package_occurrences[req["package"]].append(req)
            
        duplicates_found = False
        for package, occurrences in package_occurrences.items():
            if len(occurrences) > 1:
                duplicates_found = True
                lines = [str(occ["line_num"]) for occ in occurrences]
                versions = [occ["version_spec"] for occ in occurrences]
                
                self.log_issue(
                    "DUPLICATE_DEPENDENCY",
                    f"重复依赖 '{package}' 出现在行: {', '.join(lines)}, 版本: {versions}"
                )
                
                # 检查版本冲突
                unique_versions = set(versions)
                if len(unique_versions) > 1:
                    self.log_issue(
                        "VERSION_CONFLICT", 
                        f"包 '{package}' 存在版本冲突: {list(unique_versions)}"
                    )
                    
        if not duplicates_found:
            self.log_info("DUPLICATE_CHECK", "未发现重复依赖")

    def check_version_strategy(self, requirements: List[Dict]):
        """检查版本策略 - 问题源2"""
        print("\n🔍 检查版本策略...")
        
        pinned_count = sum(1 for req in requirements if req["is_pinned"])
        range_count = sum(1 for req in requirements if req["is_range"])
        total_count = len(requirements)
        
        pinned_ratio = pinned_count / total_count if total_count > 0 else 0
        
        self.log_info("VERSION_STRATEGY", f"版本策略统计:")
        self.log_info("VERSION_STRATEGY", f"  - 固定版本 (==): {pinned_count} ({pinned_ratio:.1%})")
        self.log_info("VERSION_STRATEGY", f"  - 版本范围 (>=,<): {range_count} ({(range_count/total_count):.1%})")
        
        if pinned_ratio > 0.8:
            self.log_warning(
                "VERSION_STRATEGY_RIGID",
                f"过多固定版本依赖 ({pinned_ratio:.1%})，可能导致依赖解析困难和兼容性问题"
            )
            
        # 检查具体的固定版本依赖
        for req in requirements:
            if req["is_pinned"]:
                self.log_info(
                    "PINNED_VERSION",
                    f"行 {req['line_num']}: {req['package']}{req['version_spec']}"
                )

    def check_production_vs_dev_dependencies(self, requirements: List[Dict]):
        """检查生产vs开发依赖混合 - 问题源3"""
        print("\n🔍 检查依赖分类...")
        
        # 常见的开发工具依赖
        dev_tools = {
            'pytest', 'pytest-cov', 'pytest-mock', 'pytest-asyncio',
            'black', 'flake8', 'mypy', 'isort', 'bandit',
            'pre-commit', 'tox', 'coverage', 'sphinx',
            'jupyter', 'notebook', 'ipython',
            'debugpy', 'pdb', 'pudb'
        }
        
        dev_deps_in_prod = []
        for req in requirements:
            if req["package"].lower() in dev_tools:
                dev_deps_in_prod.append(req)
                
        if dev_deps_in_prod:
            self.log_warning(
                "DEV_DEPS_IN_PROD",
                f"发现 {len(dev_deps_in_prod)} 个开发工具依赖混入生产环境:"
            )
            for dep in dev_deps_in_prod:
                self.log_warning(
                    "DEV_DEPENDENCY",
                    f"  行 {dep['line_num']}: {dep['package']} (应移至requirements-dev.txt)"
                )
        else:
            self.log_info("DEPENDENCY_SEPARATION", "未发现开发依赖混入生产环境")

    def check_security_vulnerabilities(self, requirements: List[Dict]):
        """检查安全漏洞 - 问题源4"""
        print("\n🔍 检查安全漏洞...")
        
        try:
            # 使用pip-audit检查安全漏洞（如果可用）
            result = subprocess.run(
                ['pip-audit', '--format=json', '--requirement', str(self.requirements_file)],
                capture_output=True, text=True, timeout=30
            )
            
            if result.returncode == 0:
                audit_data = json.loads(result.stdout)
                vulnerabilities = audit_data.get('vulnerabilities', [])
                
                if vulnerabilities:
                    self.log_issue(
                        "SECURITY_VULNERABILITIES",
                        f"发现 {len(vulnerabilities)} 个安全漏洞"
                    )
                    for vuln in vulnerabilities[:5]:  # 只显示前5个
                        package = vuln.get('package', 'unknown')
                        version = vuln.get('installed_version', 'unknown')
                        advisory = vuln.get('advisory', 'No details')
                        self.log_issue(
                            "VULNERABILITY",
                            f"{package}=={version}: {advisory}"
                        )
                else:
                    self.log_info("SECURITY_CHECK", "未发现已知安全漏洞")
            else:
                self.log_warning("SECURITY_CHECK", "pip-audit不可用，跳过安全检查")
                
        except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError):
            self.log_warning("SECURITY_CHECK", "安全检查失败，可能需要安装pip-audit")

    def check_file_structure(self, requirements: List[Dict]):
        """检查文件结构和组织 - 问题源5"""
        print("\n🔍 检查文件结构...")
        
        # 检查是否有逻辑分组（通过注释）
        with open(self.requirements_file, 'r', encoding='utf-8') as f:
            content = f.read()
            
        comment_lines = [line for line in content.split('\n') if line.strip().startswith('#')]
        section_comments = [line for line in comment_lines if any(keyword in line.lower() 
                          for keyword in ['core', 'web', 'database', 'api', 'ml', 'data', 'testing'])]
        
        if len(section_comments) < 3:
            self.log_warning(
                "FILE_ORGANIZATION",
                f"文件缺乏逻辑分组，建议按功能模块组织依赖 (当前只有 {len(section_comments)} 个分组注释)"
            )
        else:
            self.log_info("FILE_ORGANIZATION", f"发现 {len(section_comments)} 个功能分组")

    def check_platform_specific_dependencies(self, requirements: List[Dict]):
        """检查平台特定依赖 - 问题源6"""
        print("\n🔍 检查平台特定依赖...")
        
        platform_specific_packages = {
            'pywin32', 'wmi', 'winshell',  # Windows
            'pyobjc', 'pyobjc-core',       # macOS
            'python-dbus', 'pygobject'      # Linux
        }
        
        platform_deps = []
        for req in requirements:
            if req["package"].lower() in platform_specific_packages:
                platform_deps.append(req)
                
        if platform_deps:
            self.log_warning(
                "PLATFORM_SPECIFIC",
                f"发现 {len(platform_deps)} 个平台特定依赖，建议添加条件约束:"
            )
            for dep in platform_deps:
                self.log_warning(
                    "PLATFORM_DEPENDENCY",
                    f"  行 {dep['line_num']}: {dep['package']} (建议添加 ; sys_platform 条件)"
                )

    def check_dependency_completeness(self, requirements: List[Dict]):
        """检查依赖完整性 - 问题源7"""
        print("\n🔍 检查依赖完整性...")
        
        try:
            # 尝试解析依赖树
            result = subprocess.run(
                ['pip', 'check'], capture_output=True, text=True, timeout=15
            )
            
            if result.returncode != 0:
                self.log_warning(
                    "DEPENDENCY_CONFLICTS",
                    f"依赖冲突检测: {result.stdout.strip()}"
                )
            else:
                self.log_info("DEPENDENCY_CHECK", "依赖完整性检查通过")
                
        except (subprocess.TimeoutExpired, FileNotFoundError):
            self.log_warning("DEPENDENCY_CHECK", "无法执行依赖完整性检查")

    def generate_diagnosis_report(self):
        """生成诊断报告"""
        print("\n" + "="*60)
        print("📋 REQUIREMENTS.TXT 诊断报告")
        print("="*60)
        
        total_issues = len(self.issues)
        total_warnings = len(self.warnings)
        
        print(f"\n📊 问题统计:")
        print(f"  - 严重问题: {total_issues}")
        print(f"  - 警告: {total_warnings}")
        print(f"  - 信息: {len(self.info)}")
        
        if total_issues > 0:
            print(f"\n❌ 发现的严重问题:")
            for issue in self.issues:
                line_info = f" (行 {issue['line']})" if 'line' in issue else ""
                print(f"  - {issue['category']}: {issue['message']}{line_info}")
                
        if total_warnings > 0:
            print(f"\n⚠️  警告:")
            for warning in self.warnings:
                line_info = f" (行 {warning['line']})" if 'line' in warning else ""
                print(f"  - {warning['category']}: {warning['message']}{line_info}")
        
        # 问题优先级分析
        print(f"\n🎯 最可能的问题源分析:")
        
        # 统计问题类别
        issue_categories = Counter([issue['category'] for issue in self.issues])
        warning_categories = Counter([warning['category'] for warning in self.warnings])
        
        priority_issues = []
        
        if issue_categories.get('DUPLICATE_DEPENDENCY', 0) > 0:
            priority_issues.append("1. 重复依赖 - 导致版本冲突和安装失败")
            
        if warning_categories.get('VERSION_STRATEGY_RIGID', 0) > 0:
            priority_issues.append("2. 版本策略过于严格 - 限制依赖解析灵活性")
            
        if warning_categories.get('DEV_DEPS_IN_PROD', 0) > 0:
            priority_issues.append("3. 开发依赖混入生产环境 - 违反最佳实践")
            
        if priority_issues:
            for issue in priority_issues:
                print(f"  {issue}")
        else:
            print("  未发现高优先级问题")
            
        return {
            "total_issues": total_issues,
            "total_warnings": total_warnings,
            "issues": self.issues,
            "warnings": self.warnings,
            "priority_issues": priority_issues
        }

    def run_full_diagnosis(self):
        """运行完整诊断"""
        print("🔍 开始 requirements.txt 系统性诊断...")
        print(f"📁 分析文件: {self.requirements_file}")
        
        # 解析requirements文件
        requirements = self.parse_requirements()
        if not requirements:
            print("❌ 无法解析requirements文件，诊断终止")
            return None
            
        print(f"📦 发现 {len(requirements)} 个依赖包")
        
        # 执行各项检查
        self.check_duplicate_dependencies(requirements)
        self.check_version_strategy(requirements)
        self.check_production_vs_dev_dependencies(requirements)
        self.check_security_vulnerabilities(requirements)
        self.check_file_structure(requirements)
        self.check_platform_specific_dependencies(requirements)
        self.check_dependency_completeness(requirements)
        
        # 生成报告
        return self.generate_diagnosis_report()

def main():
    """主函数"""
    requirements_file = sys.argv[1] if len(sys.argv) > 1 else "requirements.txt"
    
    diagnostic = RequirementsDiagnostic(requirements_file)
    result = diagnostic.run_full_diagnosis()
    
    if result and (result["total_issues"] > 0 or result["total_warnings"] > 0):
        print(f"\n💡 建议:")
        print(f"  1. 优先解决重复依赖问题")
        print(f"  2. 考虑采用兼容版本范围策略 (>=x.y.z,<major+1.0.0)")
        print(f"  3. 将开发工具依赖移至 requirements-dev.txt")
        print(f"  4. 按功能模块重新组织文件结构")
        
        return 1 if result["total_issues"] > 0 else 0
    else:
        print(f"\n✅ requirements.txt 文件状态良好!")
        return 0

if __name__ == "__main__":
    exit(main())